package mqttx

import (
	"crypto/tls"
	"fmt"
	"log"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/pkg/errors"
	"github.com/zeromicro/go-zero/core/logx"
)

type MqttConfig struct {
	Broker      string
	User        string
	Password    string
	Certificate string `json:",optional"` // 证书文件
	PrivateKey  string `json:",optional"` // 密钥
	ClientId    string
	WillEnabled bool   `json:",optional"` // 遗嘱
	WillTopic   string `json:",optional"` // 遗嘱主题
	WillPayload string `json:",optional"` // 遗嘱消息
	WillQos     byte   `json:",optional"` // 遗嘱服务质量，目前只有三个等级：0，1，2
	Qos         byte   `json:",optional"` // 服务质量,目前只有三个等级：0，1，2
	Retained    bool   `json:",optional"` // 保留消息
}

type MqttClient struct {
	qos      byte
	retained bool
	Client   mqtt.Client

	subscriptions map[string]mqtt.MessageHandler
	mux           sync.Mutex
}

var (
	once      sync.Once
	singleton *MqttClient
)

// 证书
func newTLSConfig(certFile, privateKey string) (*tls.Config, error) {
	cert, err := tls.LoadX509KeyPair(certFile, privateKey)
	if err != nil {
		return nil, err
	}
	return &tls.Config{
		Certificates:       []tls.Certificate{cert},
		ClientAuth:         tls.NoClientCert, // 不需要证书
		ClientCAs:          nil,              // 不需要验证证书
		InsecureSkipVerify: true,             //
	}, nil
}

func NewClient(conf MqttConfig) *MqttClient {
	once.Do(func() {
		opts := mqtt.NewClientOptions().AddBroker(conf.Broker).SetClientID(conf.ClientId)
		if conf.WillEnabled {
			opts.SetWill(conf.WillTopic, conf.WillPayload, conf.WillQos, conf.Retained)
		}
		if conf.Certificate != "" {
			tlsConf, err := newTLSConfig(conf.Certificate, conf.PrivateKey)
			if err != nil {
				log.Panic(err)
				return
			}
			opts.SetTLSConfig(tlsConf)
		}
		if conf.User != "" && conf.Password != "" {
			opts.SetUsername(conf.User).SetPassword(conf.Password)
		}

		opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
			fmt.Printf("mqtt: disconnected (%v), reconnecting...\n", err)
		})

		opts.SetOnConnectHandler(func(client mqtt.Client) {
			fmt.Printf("Connect success, %v\n", client.IsConnected())
			currentSingleton := singleton
			if currentSingleton == nil {
				fmt.Println("mqtt: singleton instance is nil, cannot resubscribe")
				return
			}
			for topic, handler := range currentSingleton.subscriptions {
				fmt.Printf("mqtt: re-subscribing to topic %s\n", topic)
				if err := currentSingleton.Subscribe([]string{topic}, handler); err != nil {
					fmt.Printf("mqtt: failed to re-subscribe to topic %s\n", topic)
				}
			}
		})

		opts.SetAutoReconnect(true)
		opts.SetResumeSubs(true)
		opts.SetKeepAlive(30 * time.Second)
		opts.SetPingTimeout(3 * time.Second)
		opts.SetCleanSession(false)

		c := &MqttClient{
			Client:        mqtt.NewClient(opts),
			qos:           conf.Qos,
			retained:      conf.Retained,
			subscriptions: make(map[string]mqtt.MessageHandler),
		}
		if token := c.Client.Connect(); token.Wait() && token.Error() != nil {
			log.Panic(token.Error())
			return
		}

		singleton = c
	})

	return singleton
}

func (c *MqttClient) Publish(topic string, payload []byte) (msgId uint16, err error) {
	if c != nil && c.Client.IsConnected() {
		token := c.Client.Publish(topic, c.qos, c.retained, payload)
		if token.WaitTimeout(3 * time.Second) {
			return 0, errors.New("publish wait timeout") // should be blocked waiting for CONNACK
		}
		if token.Error() != nil { // Should never have an error
			return 0, token.Error()
		}

		// 获取消息ID
		tc, ok := token.(*mqtt.PublishToken)
		if !ok {
			return 0, nil
		}
		return tc.MessageID(), nil

	}
	return 0, errors.New("mqttclient is nil or disconnected")
}

func (c *MqttClient) Subscribe(topics []string, onMessage mqtt.MessageHandler) error {
	c.mux.Lock()
	defer c.mux.Unlock()
	for _, topic := range topics {
		c.subscriptions[topic] = onMessage
		if tc := c.Client.Subscribe(topic, c.qos, onMessage); tc.Wait() && tc.Error() != nil {
			return tc.Error()
		}
		logx.Info(fmt.Sprintf("订阅主题【%s】成功", topic))
	}
	return nil
}

func (c *MqttClient) UnSubscribe(topics []string) error {
	logx.Info(fmt.Sprintf("取消订阅主题"))
	if tc := c.Client.Unsubscribe(topics...); tc.Wait() && tc.Error() != nil {
		return tc.Error()
	}
	return nil
}

func (c *MqttClient) Close() {
	c.Client.Disconnect(250)
}
